package org.huangrui.spark.java.streaming;

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;


/**
 * nc -lp 9999
 * nc -L -p 9999
 * @Author hr
 * @Create 2024-10-21 16:16
 */
public class SparkStreaming01_WordCount {
    public static void main(String[] args) throws InterruptedException {
        // TODO 创建环境对象
        // StreamingContext创建时，需要传递两个参数
        // 第一个参数表示环境配置对象
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming");
        // 第二个参数表示批量处理的周期（采集周期），单位是毫秒
        JavaStreamingContext jsc = new JavaStreamingContext(conf, new Duration(5*1000L));

        // TODO 通过环境对象对接Socket数据源，获取数据模型，进行数据处理
        JavaReceiverInputDStream<String> socketTextStream = jsc.socketTextStream("localhost", 9999);
        JavaPairDStream<String, Integer> wordCountDS = socketTextStream.flatMap(line -> Arrays.asList(line.split(" ")).iterator())
                .mapToPair(word -> new Tuple2<>(word, 1)).reduceByKey(Integer::sum);
        wordCountDS.print();

        // TODO 启动数据采集器
        jsc.start();
        // TODO 等待数据采集器的结束，如果采集器停止运行，main线程会继续执行
        jsc.awaitTermination();
    }
}
